package com.shujia.jinjie;

import com.shujia.util.HBaseTool;
import org.apache.hadoop.hbase.CompareOperator;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.*;
import org.apache.hadoop.hbase.regionserver.BloomType;
import org.apache.hadoop.hbase.util.Bytes;

/*
    比较过滤器：
        行键过滤器 RowFilter
        列簇过滤器 FamilyFilter
        列名过滤器 QualifierFilter
        列值过滤器 ValueFilter
 */
public class FilterDemo {
    private static Connection conn = HBaseTool.HBASE_CONN;
    private static Admin admin = HBaseTool.HBASE_ADMIN;

    public static void main(String[] args) {
        //通过RowFilter与BinaryComparator过滤比rowKey 1500100010小的所有值出来
//        fun1(); //行键过滤器 RowFilter

        //通过FamilyFilter与SubstringComparator查询列簇名包含i的所有列簇下面的数据
//        fun2();

        //通过QualifierFilter与SubstringComparator查询列名包含 ‘ge’ 的列的值
//        fun3();

        //通过ValueFilter与BinaryPrefixComparator过滤出所有的cell中值以 "张" 开头的学生
        //ValueFilter: 只要列值符合条件就过滤出来，不符合不会查询出来
//        fun4();
//
//        System.out.println("======================================");
        //SingleColumnValueFilter 单列值过滤器 "张" 开头的学生
        //SingleColumnValueFilter可以指定过滤的列值是哪一列的。
//        fun5();

        /**
         *  ValueFilter与SingleColumnValueFilter的区别：
         *  1、ValueFilter是在所有列的范围内查找符合条件的列，结果只会有符合条件的列，创建时不需要指定列名
         *  2、SingleColumnValueFilter是在指定的列名中查询符合的列，结果除了符合条件列以外，与该列所属同一行的其他列也会获取，创建时需要指定过滤哪一列的列值
         *
         */
        //需求：查询理科一班，年龄大于22岁的学生
//        fun6();

//        System.out.println("------------------------");
        // 使用SingleColumnValueExcludeFilter 单列值排除过滤器查询姓名以"张" 开头的学生
//        fun7();


        // 需求过滤出学号是以150010098 开头的学生
        //使用上面的比较过滤器中的行键过滤器，要配合操作符和前缀比较器来实现过滤
//        System.out.println("使用比较过滤器中的 RowFilter");
//        fun8();
//        System.out.println("-----------------------------");
//        System.out.println("使用专用过滤器中的 PrefixFilter");
//        //使用专用过滤器中的PrefixFilter
//        fun9();

        //需求：分页查询前三页数据，每一页为10条数据
//        fun10();

        //需求：使用跳过过滤器
//        fun11();

        //创建一张表并设置布隆过滤器，布隆过滤器是作用在每个文件上的，每个文件代表一个列簇部分的数据
        fun12();
    }

    public static void fun12(){
        try {
            TableDescriptorBuilder fun12 = TableDescriptorBuilder.newBuilder(TableName.valueOf("fun13"));

//            ColumnFamilyDescriptor info = ColumnFamilyDescriptorBuilder.of("info");
            //针对列簇设置布隆过滤器 老版本默认不开启布隆过滤器，新版本列簇默认是ROW级别的过滤器
            ColumnFamilyDescriptor info = ColumnFamilyDescriptorBuilder
                    .newBuilder(Bytes.toBytes("info"))
                    .setBloomFilterType(BloomType.ROWCOL)
                    .build();

            //将列簇添加到表中
            fun12.setColumnFamily(info);

            Admin hbaseAdmin = HBaseTool.HBASE_ADMIN;
            hbaseAdmin.createTable(fun12.build());


        }catch (Exception e){
            e.printStackTrace();
        }
    }

    public static void fun11(){
        try {
            Table students = HBaseTool.getOneTable("students");
            Scan scan = new Scan();

            Filter filter = new ValueFilter(CompareOperator.NOT_EQUAL, new BinaryComparator(Bytes.toBytes("文科一班")));

            SkipFilter skipFilter = new SkipFilter(filter);

//            scan.setFilter(filter);
//
            scan.setFilter(skipFilter);

            ResultScanner resultScanner = students.getScanner(scan);
            HBaseTool.printMoreData(resultScanner);

        }catch (Exception e){
            e.printStackTrace();
        }
    }

    public static void fun10() {
        try {
            Table students = HBaseTool.getOneTable("students");

            //定义页数
            int pageNum = 3;

            //定义每一页的数据条数
            int pageDataNum = 10;

            //先指定开始行键
            String startRowKey = "";

            //创建一个分页过滤器
            PageFilter pageFilter = new PageFilter(pageDataNum);

            for (int i = 1; i <= 3; i++) {
                System.out.println("==========================当前是第 " + i + " 页==========================");
                Scan scan = new Scan();
                scan.withStartRow(Bytes.toBytes(startRowKey));
                scan.setFilter(pageFilter);
                ResultScanner resultScanner = students.getScanner(scan);
                for (Result result : resultScanner) {
                    HBaseTool.printOneData(result);
                    startRowKey = Bytes.toString(result.getRow()) + " ";
                }
            }


        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public static void fun9() {
        try {
            Table students = HBaseTool.getOneTable("students");
            Scan scan = new Scan();
            //创建行键前缀过滤器
            PrefixFilter prefixFilter = new PrefixFilter(Bytes.toBytes("150010098"));
            scan.setFilter(prefixFilter);
            ResultScanner resultScanner = students.getScanner(scan);
            HBaseTool.printMoreData(resultScanner);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public static void fun8() {
        try {
            Table students = HBaseTool.getOneTable("students");
            Scan scan = new Scan();
            //RowFilter(final CompareOperator op, final ByteArrayComparable rowComparator)
            //创建等值比较器
            BinaryPrefixComparator binaryPrefixComparator = new BinaryPrefixComparator(Bytes.toBytes("150010098"));
            RowFilter rowFilter = new RowFilter(CompareOperator.EQUAL, binaryPrefixComparator);

            //Scan setFilter(Filter filter) 向Scan对象中设置过滤器，这个方法中测试的是行键过滤器
            scan.setFilter(rowFilter);
            ResultScanner resultScanner = students.getScanner(scan);
            HBaseTool.printMoreData(resultScanner);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }


    public static void fun7() {
        try {
            //获取表对象
            Table students = HBaseTool.getOneTable("students");
            Scan scan = new Scan();
            //创建单列值排除过滤器
            SingleColumnValueExcludeFilter singleColumnValueExcludeFilter = new SingleColumnValueExcludeFilter(
                    Bytes.toBytes("info"),
                    Bytes.toBytes("name"),
                    CompareOperator.EQUAL,
                    new BinaryPrefixComparator(Bytes.toBytes("张"))
            );

            scan.setFilter(singleColumnValueExcludeFilter);

            ResultScanner resultScanner = students.getScanner(scan);
            HBaseTool.printMoreData(resultScanner);


        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public static void fun6() {
        try {
            Table students = HBaseTool.getOneTable("students");
            Scan scan = new Scan();
            //创建单列值过滤器
            //public SingleColumnValueFilter(byte[] family, byte[] qualifier, CompareOperator op, ByteArrayComparable comparator)
            BinaryComparator binaryComparator = new BinaryComparator(Bytes.toBytes("理科一班"));
            SingleColumnValueFilter singleColumnValueFilter = new SingleColumnValueFilter(
                    Bytes.toBytes("info"),
                    Bytes.toBytes("clazz"),
                    CompareOperator.EQUAL,
                    binaryComparator);

            //创建单列值过滤器
            //public SingleColumnValueFilter(byte[] family, byte[] qualifier, CompareOperator op, ByteArrayComparable comparator)
            BinaryComparator binaryComparator2 = new BinaryComparator(Bytes.toBytes("22"));
            SingleColumnValueFilter singleColumnValueFilter2 = new SingleColumnValueFilter(
                    Bytes.toBytes("info"),
                    Bytes.toBytes("age"),
                    CompareOperator.GREATER,
                    binaryComparator2);

            //创建单列值过滤器
            //public SingleColumnValueFilter(byte[] family, byte[] qualifier, CompareOperator op, ByteArrayComparable comparator)
            BinaryComparator binaryComparator3 = new BinaryComparator(Bytes.toBytes("女"));
            SingleColumnValueFilter singleColumnValueFilter3 = new SingleColumnValueFilter(
                    Bytes.toBytes("info"),
                    Bytes.toBytes("gender"),
                    CompareOperator.EQUAL,
                    binaryComparator3);

            FilterList filterList = new FilterList();
            filterList.addFilter(singleColumnValueFilter);
            filterList.addFilter(singleColumnValueFilter2);
            filterList.addFilter(singleColumnValueFilter3);

            scan.setFilter(filterList);


            ResultScanner resultScanner = students.getScanner(scan);
            HBaseTool.printMoreData(resultScanner);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }


    public static void fun5() {
        try {
            Table students = HBaseTool.getOneTable("students");
            Scan scan = new Scan();
            //创建单列值过滤器
            //public SingleColumnValueFilter(byte[] family, byte[] qualifier, CompareOperator op, ByteArrayComparable comparator)
            BinaryPrefixComparator binaryPrefixComparator = new BinaryPrefixComparator(Bytes.toBytes("张"));
            SingleColumnValueFilter singleColumnValueFilter = new SingleColumnValueFilter(
                    Bytes.toBytes("info"),
                    Bytes.toBytes("name"),
                    CompareOperator.EQUAL,
                    binaryPrefixComparator);

            scan.setFilter(singleColumnValueFilter);

            ResultScanner resultScanner = students.getScanner(scan);
            HBaseTool.printMoreData(resultScanner);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }


    public static void fun4() {
        try {
            Table students = HBaseTool.getOneTable("students");
            Scan scan = new Scan();
            //创建二进制前缀比较器
            BinaryPrefixComparator binaryPrefixComparator = new BinaryPrefixComparator(Bytes.toBytes("张"));

            //构造一个列值过滤器,获取所有符合条件的列，不符合条件的就不会获取
            ValueFilter valueFilter = new ValueFilter(CompareOperator.EQUAL, binaryPrefixComparator);
            scan.setFilter(valueFilter);

            ResultScanner resultScanner = students.getScanner(scan);
            HBaseTool.printMoreData(resultScanner);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public static void fun3() {
        try {
            Table students = HBaseTool.getOneTable("students");
            Scan scan = new Scan();
            //创建包含比较器
            SubstringComparator substringComparator = new SubstringComparator("ge");
            //创建列名过滤器
            QualifierFilter qualifierFilter = new QualifierFilter(CompareOperator.EQUAL, substringComparator);
            scan.setFilter(qualifierFilter);

            ResultScanner resultScanner = students.getScanner(scan);
            HBaseTool.printMoreData(resultScanner);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }


    public static void fun2() {
        try {
            Table users = HBaseTool.getOneTable("users");
            Scan scan = new Scan();
            //FamilyFilter(final CompareOperator op, final ByteArrayComparable familyComparator)
            SubstringComparator substringComparator = new SubstringComparator("nf");
            //创建列簇过滤器
            FamilyFilter familyFilter = new FamilyFilter(CompareOperator.EQUAL, substringComparator);
            scan.setFilter(familyFilter);
            ResultScanner resultScanner = users.getScanner(scan);
            HBaseTool.printMoreData(resultScanner);

        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public static void fun1() {
        try {
            Table students = HBaseTool.getOneTable("students");
            Scan scan = new Scan();
            //RowFilter(final CompareOperator op, final ByteArrayComparable rowComparator)
            //创建等值比较器
            BinaryComparator binaryComparator = new BinaryComparator(Bytes.toBytes("1500100020"));
            RowFilter rowFilter = new RowFilter(CompareOperator.LESS, binaryComparator);

            //Scan setFilter(Filter filter) 向Scan对象中设置过滤器，这个方法中测试的是行键过滤器
            scan.setFilter(rowFilter);
            ResultScanner resultScanner = students.getScanner(scan);
            HBaseTool.printMoreData(resultScanner);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
